package com.kingsoft.dc.khaos.module.spark.preprocess.specific

import java.util.Properties

import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.metadata.Dependency
import com.kingsoft.dc.khaos.module.spark.constants.RedisConstants
import com.kingsoft.dc.khaos.module.spark.preprocess.transform.TransformStrategy
import com.kingsoft.dc.khaos.module.spark.util.RedisUtils
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.DataFrame
import redis.clients.jedis.{Jedis, JedisCluster, JedisPool, JedisSentinelPool}

import scala.collection.mutable

/**
  * create by haorenhui on 2020/06/28.
  */
class ClearRedisSyncData extends TransformStrategy with Logging {

  override def exec(kc: KhaosContext,
                    module_id: String,
                    config: String,
                    dependences: Seq[Dependency],
                    targets: Seq[Dependency]): Seq[(String, DataFrame)] = {
    val idClearSyncData: String = kc._valuesChannel.getValues[String](RedisConstants.isClearSyncData).toLowerCase.trim
    if(idClearSyncData.toBoolean){
      val connInfo: mutable.HashMap[String, String] = kc._valuesChannel.getValues[mutable.HashMap[String, String]](RedisConstants.clearSyncDataInfo)
      val connectType: String = connInfo(RedisConstants.clearSyncDataInfoConnectType)
      val redisNodes: String = connInfo(RedisConstants.clearSyncDataInfoRedisNodes)
      val redisPassword: String = connInfo.getOrElse(RedisConstants.clearSyncDataInfoRedisPassword,null)
      val masterName: String = connInfo.getOrElse(RedisConstants.clearSyncDataInfoMasterName,null)

      val keys: Array[String] = kc._valuesChannel.getValues[Array[String]](RedisConstants.clearSyncDataInfoKeys)
      if(keys.isEmpty){
        log.info("redis无数据删除")
      }else{
        val redisProperties: Map[String, String] = kc.conf.getAllWithPrefix("module.redis.delete.").toMap
        log.info("redisClear redisProperties")
        redisProperties.foreach(x=>log.info(x._1 + "   " + x._2))
        val deleteLength: Int = redisProperties.getOrElse(RedisConstants.DELETE_PER_DELETE_NUMS,RedisConstants.DEFAULT_DELETE_PER_DELETE_NUMS).toInt
        log.info(s"redis ClearRedisSyncData redisDeleteLength :$deleteLength")
        idClearSyncData match {
          case "false" => //do nothing
          case "true" =>
            connectType match {
              case RedisConstants.RedisConnectEnum.cluster =>
                val conn:JedisCluster = RedisUtils.getClusterPool(redisNodes,redisPassword, redisProperties,this)
                val keysList: mutable.Buffer[mutable.Buffer[String]] = getBatchKeys(keys,deleteLength)
                keysList.foreach(batchKeys=>
                  RedisUtils.deleteDataByKeys(conn,batchKeys.toArray)
                )
                RedisUtils.closeConn(conn)
              case RedisConstants.RedisConnectEnum.sentinel =>
                val pool: JedisSentinelPool = RedisUtils.getSentinelPool(redisNodes,redisPassword,masterName, redisProperties,this)
                val conn:Jedis = pool.getResource
                val keysList: mutable.Buffer[mutable.Buffer[String]] = getBatchKeys(keys,deleteLength)
                keysList.foreach(batchKeys=>
                  RedisUtils.deleteDataByKeys(conn,batchKeys.toArray)
                )
                RedisUtils.closeConn(conn)
                RedisUtils.closeConnPool(pool)
              case RedisConstants.RedisConnectEnum.masterSlave =>
                val pool: JedisPool = RedisUtils.getSinglePool(redisNodes,redisPassword, redisProperties,this)
                val conn:Jedis = pool.getResource
                val keysList: mutable.Buffer[mutable.Buffer[String]] = getBatchKeys(keys,deleteLength)
                keysList.foreach(batchKeys=>
                  RedisUtils.deleteDataByKeys(conn,batchKeys.toArray)
                )
                RedisUtils.closeConn(conn)
                RedisUtils.closeConnPool(pool)
              case _ => throw new Exception(s"不支持的redis类型 $connectType")
            }

            log.info("redis已同步的数据删除成功")
          case _ => // do nothing
        }
      }

    }

    val df: DataFrame = null
    addResult(Dependency(), df)
  }

  //分批提交删除命令
  def getBatchKeys(keys: Array[String],deleteLength:Int): mutable.Buffer[mutable.Buffer[String]] ={
    val keysList:mutable.Buffer[mutable.Buffer[String]]=mutable.Buffer[mutable.Buffer[String]]()
    var arrBuffer:mutable.Buffer[String]=mutable.Buffer[String]()
    for(i <- 0 until keys.length){
      if(arrBuffer.length != deleteLength){
        arrBuffer.append(keys(i))
      }else{
        keysList.append(arrBuffer)
        arrBuffer= mutable.Buffer[String]()
        arrBuffer.append(keys(i))
      }
    }
    if(arrBuffer.nonEmpty)
      keysList.append(arrBuffer)
    keysList
  }

  override def schema(kc: KhaosContext,
                      module_id: String,
                      config: String,
                      dependences: Seq[Dependency]): Any = {
    new Schema(Nil)
  }
}
